package com.binance.notification.web.service.impl;

import java.sql.Date;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import com.binance.master.utils.DateUtils;
import com.binance.notification.api.push.PushCommand;
import com.binance.notification.api.push.notification.AbstractNotification;
import com.binance.notification.api.push.notification.SimpleNotification;
import com.binance.notification.api.push.notification.SingleNotificationPacket;
import com.binance.notification.web.push.PushResult;
import com.binance.notification.web.push.RequestCallback;
import com.binance.notification.web.service.IAlertTradeService;
import com.binance.notification.web.service.IPushService;
import com.binance.notification.web.utils.Worker;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;

@Service
@Slf4j
public class AlertTradeServiceImpl implements IAlertTradeService {

    private static final LinkedBlockingQueue<String> ALERT_TRADE_QUEUE = new LinkedBlockingQueue<>();
    @Autowired
    private IPushService pushService;
    @Value("${jpush.alert.trade.batch.count:1000}")
    private int batchCount;
    @Value("${jpush.alert.trade.enabled}")
    private boolean alertTradeEnabled;
    @Resource
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private Worker worker;

    @Override
    public void pushForAlert(String tradeStr) {
        if (!alertTradeEnabled) {
            throw new UnsupportedOperationException(
                    "jpush.alert.trade.enabled must to be 'true' to enable the operation.");
        }
        ALERT_TRADE_QUEUE.add(tradeStr);
    }

    @PostConstruct
    public void init() {
        for (int i = 0; i < 5; i++) {
            worker.submit(new Runnable() {

                @Override
                public void run() {
                    pushForAlert();
                }

            });
        }
    }

    private void pushForAlert() {
        PushBody body = new PushBody();
        while (alertTradeEnabled) {
            try {
                String tradeStr = null;
                if (body.size() == 0) {
                    try {
                        tradeStr = ALERT_TRADE_QUEUE.take();
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_TRADE_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                } else {
                    try {
                        tradeStr = ALERT_TRADE_QUEUE.poll(50, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        log.warn("等待从ALERT_TRADE_QUEUE获取元素出现异常", e);
                        Thread.currentThread().interrupt();
                    }
                }
                if (tradeStr == null) {
                    try {
                        this.doPush(body);
                    } finally {
                        body = new PushBody();
                    }
                    continue;
                } else {
                    body.addPayload(tradeStr);
                }
                if (this.batchCount > 0) {
                    // 支持批处理
                    if (body.size() == this.batchCount) {
                        try {
                            this.doPush(body);
                        } finally {
                            body = new PushBody();
                        }
                    }
                } else {
                    // 不支持批处理，则立即推送
                    try {
                        this.doPush(body);
                    } finally {
                        body = new PushBody();
                    }
                }
            } catch (Exception e) {
                log.error("pushForAlert()出现异常", e);
                body = new PushBody();
            }

        }
    }

    private void doPush(PushBody body) {
        if (body.size() > 0) {
            this.pushService.sendSingle(body.pushPayload, new RequestCallback() {

                @Override
                public void completed(List<PushResult> resultList, Object... obj) {
                    onPush(resultList, body);
                }

            });
        }
    }

    private void onPush(List<PushResult> pushResultList, PushBody body) {
        for (PushResult pushResult : pushResultList) {
            if (pushResult.isResultOK()) {

            } else {
                if (pushResult.getEx() != null) {
                    log.error("发送出现异常", pushResult.getEx());
                } else {
                    log.error("发送失败：{}", pushResult.getOriginalContent());
                }
            }
        }
    }

    private SingleNotificationPacket<SimpleNotification> createPushPayload(String alertTrade) {
        String[] array = StringUtils.split(alertTrade, ",");
        String userId = array[0];
        String pair = array[1];
        // String type = array[2];// 0 买，1 卖
        String price = array[3];
        String amount = array[4];
        String status = array[5]; // 0 部分成交，1 完全成交
        String date = array[6];
        String orderId = array[7];

        if (!this.canPush(orderId, status)) {
            return null;
        }
        String content = "";
        if ("0".equals(status)) {
            if ("0".equals(status)) {
                content = "#{trade.tip0:" + pair + "," + price + "," + amount + "}";
            } else {
                content = "#{trade.tip2:" + pair + "," + price + "," + amount + "}";
            }
        } else {
            if ("0".equals(status)) {
                DateUtils.formatter(new Date(Long.parseLong(date)), "yyyy-MM-dd HH:mm:ss");
                content = "#{trade.tip1:" + pair + "," + price + "," + amount + "}";
            } else {
                content = "#{trade.tip3:" + pair + "," + price + "," + amount + "}";
            }
        }

        SimpleNotification simpleNotification = new SimpleNotification();
        simpleNotification.setTitle("");
        simpleNotification.setContent(content);

        SingleNotificationPacket<SimpleNotification> packet = new SingleNotificationPacket<>();
        packet.setMessage(simpleNotification);
        PushCommand command = new PushCommand();
        command.setTagNotList(Arrays.asList("TN"));
        command.getUsers().add(userId);
        packet.setCommand(command);
        return packet;
    }

    private boolean canPush(String orderId, String status) {
        if ("0".equals(status)) {
            return this.redisTemplate.opsForValue().setIfAbsent(orderId + ",0", "0");
        }
        // 当全部成交时删除redis中的key
        if ("1".equals(status)) {
            this.redisTemplate.delete(orderId + ",0");
        }
        return true;
    }

    private class PushBody {
        List<SingleNotificationPacket<? extends AbstractNotification>> pushPayload = Lists.newArrayList();

        public void addPayload(String alertTrade) {
            SingleNotificationPacket<SimpleNotification> packet = createPushPayload(alertTrade);
            if (packet != null) {
                this.pushPayload.add(packet);
            }
        }

        public int size() {
            return this.pushPayload.size();
        }
    }

}
